package com.gao.flying.mq.core;

import java.util.Properties;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

import com.gao.flying.mq.config.Broker;
import com.gao.flying.mq.connection.CachingConnectionFactory;
import com.gao.flying.mq.connection.JmsResource;
import com.gao.flying.mq.support.MessageConverter;
import com.gao.flying.mq.support.SimpleMessageConverter;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.commons.lang3.StringUtils;


public class JmsContext {
    private ConnectionFactory targetConnectionFactory = null;
    private CachingConnectionFactory cachingConnectionFactory = null;
    private MessageConverter messageConverter = null;

    private JmsResource resource = new JmsResource();

    private Broker broker;

    public JmsContext(Broker broker) {
        this.broker = broker;
        targetConnectionFactory = new ActiveMQConnectionFactory(broker.getUrl());
        cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory.setSessionCacheSize(500);
        cachingConnectionFactory.setTargetConnectionFactory(targetConnectionFactory);
    }

    public CachingConnectionFactory getConnectionFactory() {
        return cachingConnectionFactory;
    }

    public Connection getConnection() throws JMSException {
        Connection connection = resource.getConnection();
        if (connection == null) {
            if (StringUtils.isAnyBlank(broker.getUsername(), broker.getPassword())) {
                connection = getConnectionFactory().createConnection();
            } else {
                connection = getConnectionFactory().createConnection(broker.getUsername(), broker.getPassword());
            }
            connection.start();
            resource.addConnection(connection);
        }
        return connection;
    }

    public MessageConverter getMessageConverter() {
        if (messageConverter == null) {
            messageConverter = new SimpleMessageConverter();
        }

        return messageConverter;
    }

    public Message toMessage(Object message, Session session) throws JMSException {
        return getMessageConverter().toMessage(message, session);
    }

    public Session getSession() throws JMSException {
        Connection conn;
        Session session;

        session = resource.getSession(Session.class);
        if (session == null) {
            conn = getConnection();
            session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
            resource.addSession(session, conn);
        }
        return session;
    }

    public void close() {
        resource.closeAll();
        getConnectionFactory().destroy();
    }

}
